package com.yunhe.transfer.util;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.model.authority.Station;
import com.yunhe.common.util.StringUtil;
import com.yunhe.common.util.TokenUtil;
import com.yunhe.transfer.client.AuthorityClient;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.model.authority.StationDTO;
import com.yunhe.common.model.base.DataTransferDTO;
import com.yunhe.transfer.websocket.RecollectHandler;
import com.yunhe.common.util.CollectionUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.*;

@Component
public class WebsocketUtil {

    @Autowired
    private AuthorityClient authorityClient;
    @Autowired
    private RecollectHandler recollectHandler;

    private static final ObjectMapper MAPPER = new ObjectMapper();

    public String getStationCode(String query) {
        String token = getToken(query);
        Map<String, String> map = TokenUtil.parseToken(token);
        String userId = map.get("userId");
        List<Station> stations = authorityClient.getStations(Long.parseLong(userId)).getResults();
        if (stations != null && stations.size() > 0) {
            Long code = stations.get(0).getId();
            return code != null ? String.valueOf(code) : userId;
        }
        return userId;
    }

    public List<String> getStationCodeList(String query) {
        String token = getToken(query);
        Map<String, String> map = TokenUtil.parseToken(token);
        String userId = map.get("userId");
        List<Station> stations = authorityClient.getStations(Long.parseLong(userId)).getResults();
        List<String> stationCodes = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(stations)) {
            stations.stream().forEach(stationDTO -> {
                String code = stationDTO.getId() != null ? String.valueOf(stationDTO.getCode()) : userId;
                stationCodes.add(code);
            });
            stationCodes.removeIf(Objects::isNull);
            return stationCodes;
        }
        return Collections.singletonList(userId);
    }

    public Map<String, String> getQueryParams(String query) {
        Map<String, String> params = new HashMap<>();
        if (query != null) {
            String[] arr = query.split("&");
            for (String str : arr) {
                String[] brr = str.split("=");
                if (brr.length == 2) {
                    params.put(brr[0], brr[1]);
                }
            }
        }
        return params;
    }

    public String getToken(String query) {
        return getQueryParams(query).get("access-token");
    }

    /**
     * 因socket类中无法实现kafka接受，故放置第三方类接收信息调用
     * @param record
     * @throws IOException
     */
    @KafkaListener(topics = KafkaConsts.Topic.RecollectOrder, groupId = "websocketRecollectOrder-test1")
    private void recollectTopicListener(ConsumerRecord<?, String> record) throws IOException {
        String value = record.value();
        if (!StringUtil.isEmpty(value)) {
            DataTransferDTO dataMap = MAPPER.readValue(value, DataTransferDTO.class);
            recollectHandler.recollectTopicListenerTest(dataMap);
        }
    }
}
